package com.shujia.flink.window;

import com.shujia.flink.event.MyEvent;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;

public class Demo01TimeWindow {
    public static void main(String[] args) throws Exception {
        /*
         * 时间窗口：滚动、滑动
         * 时间类型：处理时间、事件时间
         */
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        DataStream<MyEvent> myDS = env.socketTextStream("master", 8888)
                .map(new MapFunction<String, MyEvent>() {
                    @Override
                    public MyEvent map(String value) throws Exception {
                        String[] split = value.split(",");
                        return new MyEvent(split[0], Long.parseLong(split[1]));
                    }
                });

        // 基于处理时间的滚动、滑动窗口
        SingleOutputStreamOperator<Tuple2<String, Integer>> processDS = myDS.map(e -> Tuple2.of(e.getWord(), 1), Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(t2 -> t2.f0)
                // 滚动窗口 每隔5s统计一次
//                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                // 滑动窗口 每隔5s统计最近10s内的数据
                .window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(5)))
                .sum(1);

        // 基于时间时间的滚动、滑动窗口
        SingleOutputStreamOperator<MyEvent> assDS = myDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<MyEvent>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner((event, ts) -> event.getTs())
        );
        SingleOutputStreamOperator<Tuple2<String, Integer>> eventDS = assDS.map(e -> Tuple2.of(e.getWord(), 1), Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(t2 -> t2.f0)
                // 滚动窗口，由于水位线前移了5s，整体有5s的延时
//                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                // 滑动窗口
                .window(SlidingEventTimeWindows.of(Time.seconds(10),Time.seconds(5)))
                .sum(1);

//        processDS.print();
        eventDS.print();
        env.execute();

    }
}
